package com.sync.mutiParse;

import java.util.concurrent.atomic.AtomicInteger;

import com.constants.Column;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.middleware.race.sync.Server;
import com.constants.Constants;

/**
 * PaserTask是对记录的数字化转化
 */
public final class PaserTask extends Thread {
	private static final Logger loggerServer = LoggerFactory
			.getLogger(Server.class);

	public static final AtomicInteger done = new AtomicInteger(0);// 统计多少个paserTask完成
	public static int maxCount = 0; // 统计每个块最多有多少条记录
	public FreeQueue freeQueue;
	private byte[] dataBytes;
	public int inToOut = 0;
	public int outToInt = 0;
	public int outToOut = 0;
	public int intToint = 0;
	public int pkUpdateCnt = 0;
	private boolean bad;
	public PaserTask(byte[] dataBytes, boolean bad) {
		this.dataBytes = dataBytes;
		this.bad = bad;
	}

	@Override
	public void run() {
		freeQueue = new FreeQueue();
		try {
			while (true) {
				Block block = freeQueue.takeFree();
				if (ReadManager.readBlock(block) == null) {
					loggerServer
							.info("Max records Count in 64k block:" + maxCount);
					loggerServer.info(
							"inToOut:{}, outToInt:{}, outToOut:{}, intToint:{},pkUpdateCnt:{} ",
							inToOut, outToInt, outToOut, intToint, pkUpdateCnt);
					// 完成一个加一，全部完成，告诉重放线程生成完了
					if (done.incrementAndGet() == Constants.PaserThreadNum) {
						ReplayTask.producing = false;
					}
					return;
				}
				// 计算块内所有记录的主键，变更后主键，类型，地址
                int cnt;
                if (bad) {
                    cnt = computeBlock2(block);
                } else {
                    cnt = computeBlock(block);
                }

				block.recordCount = cnt;
				// 块计算完了，加入到重放队列，给重放线程读取重放，重放线程都读过后，这个块放到所属的空闲池
				ReplayManager.replayMap.put(block.id, block);
				if (cnt > maxCount) {
					maxCount = cnt;
				}
			}
		} catch (Exception e) {
			e.printStackTrace();
		}

	}

	/**
	 * 只计算主键,记录保存地址
	 * 
	 * @param block
	 * @return
	 */
	public int computeBlock(Block block) {
		// loggerServer.info(block.id + ":" + Thread.currentThread().getName());
		byte[] bytesSrc = block.buffer;
		int curPKs[] = block.curPKs;
		int changePKs[] = block.changePKs;
		int[] pos = block.pos;
		byte[] types = block.types;// 每条增量信息保存操作类型
		int i = 0;// 块内指针
		int limit = block.limit;// 块内从文件读取的字节数
		int cnt = 0;// 一共多少条增量记录的计数
		// 这里对读的块进行多线程计算主键，不重放记录，只计算，重放给重放线程
		int curPK = 0;
		int changePK = 0;
		while (i != limit) {
			i += 20;// 跳过时间戳
			while (bytesSrc[i++] != Constants.SEP) {
			}
			i += 34;// 跳过最初的timestamp和表，库
			byte tmpByte = bytesSrc[i];// 获取操作类型
			types[cnt] = tmpByte;// 保存操作的记录
			i += 2;// 跳过 U|
			// System.err.println((char) tmpByte);
			// 更新数量最多
			if (tmpByte == Constants.UPDATE) {
				// id:1:1|1265277|1265277|score:1:0|65|221|
				// 跳过id:1:1|
				i += 7;
				// 1265277|1265277|score:1:0|65|221|
				/* 获取旧主键值 */
				curPK = 0;
				while ((tmpByte = bytesSrc[i++]) != Constants.SEP) {
					curPK = curPK * 10 + tmpByte - Constants.ZERO;
				}
				/*------当前键在范围外，可以过滤--------*/
				if (curPK < Constants.start || curPK > Constants.end) {
                    while (bytesSrc[i++] != Constants.SEP) {}
                    while (( tmpByte = bytesSrc[i]) != Constants.SEPLINE) {
                        if (tmpByte == 102) {// first_name:1:0| 102
                            i += Column.COL$_LEN[0];
                        } else if (tmpByte == 108) {// last_name:1:0|
                            // 108
                            i += Column.COL$_LEN[1];
                        } else {
                            if (bytesSrc[i + 1] == 101) {// sex:1:0|
                                // 105
                                // 101
                                i += Column.COL$_LEN[2];
                            } else {
                                if (bytesSrc[i + Column.COL_LEN[3]] == Constants.SEP2) {// score:1:0|
                                    i += Column.COL$_LEN[3];
                                } else if (bytesSrc[i + Column.COL_LEN[4]] == Constants.SEP2) {// score2:1:0|
                                    i += Column.COL$_LEN[4];
                                } else {
                                    throw new RuntimeException();
                                }
                            }
                        }

                        while (bytesSrc[i++] != Constants.SEP) {} // 旧值跳过

                        while ((tmpByte = bytesSrc[i++]) != Constants.SEP) {// 读取新值
                        }
                    }
                    i++;
					continue;
				}
				changePK = 0;
				/*-----范围内主键---------*/
				while ((tmpByte = bytesSrc[i++]) != Constants.SEP) {
					changePK = changePK * 10 + tmpByte - Constants.ZERO;
				}
				if (curPK != changePK) {
					if (changePK < Constants.start
							|| changePK > Constants.end) {
					} else {
						dataBytes[changePK - Constants.start] = 1;
					}
				}
				curPKs[cnt] = curPK;
				changePKs[cnt] = changePK;
				pos[cnt] = i;
				while ((bytesSrc[i++]) != Constants.SEPLINE) {
				}
				// 插入数量其次
				cnt++;// 读取下一条记录，统计这个块里面有多少变更记录
				continue;
			} else if (tmpByte == Constants.INSERT) {
				// 跳过id:1:1|NULL|
				i += 12;
				// 1|first_name:2:0|NULL|徐|last_name:2:0|NULL|依|sex:2:0|NULL|男|score:1:0|NULL|66|score1:1:0|
				/* 计算主键值 */
				curPK = 0;
				while ((tmpByte = bytesSrc[i++]) != Constants.SEP) {
					curPK = curPK * 10 + (tmpByte - Constants.ZERO);
				}

				/*--------------*/
				if (curPK < Constants.start || curPK > Constants.end) {
					i += 75; // 主键外，不需要处理
					while ((bytesSrc[i++]) != Constants.SEPLINE) {
					}
					continue;
				}
				/*--------------*/
				dataBytes[curPK - Constants.start] = 1;
				curPKs[cnt] = curPK;
				pos[cnt] = i;// first_name:2:0|NULL|徐|last_name:2:0|NULL|依|sex:2:0|NULL|男|score:1:0|NULL|66|
				i += 75;// 后续可以进行调整
				while ((bytesSrc[i++]) != Constants.SEPLINE) {
				}
				cnt++;// 读取下一条记录，统计这个块里面有多少变更记录
				continue;
			} else {
				i += 7;
				curPK = 0;
				while ((tmpByte = bytesSrc[i++]) != Constants.SEP) {
					curPK = curPK * 10 + tmpByte - Constants.ZERO;
				}
				/*--------------*/
				if (curPK < Constants.start || curPK > Constants.end) {
					i += 75;
					while ((bytesSrc[i++]) != Constants.SEPLINE) {
					}
					continue;
				}
				/*--------------*/
				curPKs[cnt] = curPK;
				/* 删除可以直接跳跃一个大前缀 */
				i += 75;// 后续调整大小
				while ((bytesSrc[i++]) != Constants.SEPLINE) {
				}
				cnt++;// 读取下一条记录，统计这个块里面有多少变更记录
				continue;
			}

		}
		return cnt;
	}

    /**
     * 只计算主键,记录保存地址
     *
     * @param block
     * @return
     */
    public int computeBlock2(Block block) {
        // loggerServer.info(block.id + ":" + Thread.currentThread().getName());
        byte[] bytesSrc = block.buffer;
        int curPKs[] = block.curPKs;
        int changePKs[] = block.changePKs;
        int[] pos = block.pos;
        byte[] types = block.types;// 每条增量信息保存操作类型
        int i = 0;// 块内指针
        int limit = block.limit;// 块内从文件读取的字节数
        int cnt = 0;// 一共多少条增量记录的计数
        // 这里对读的块进行多线程计算主键，不重放记录，只计算，重放给重放线程
        int curPK = 0;
        int changePK = 0;
        while (i != limit) {
            i += 20;// 跳过时间戳
            while (bytesSrc[i++] != Constants.SEP) {
            }
            i += 34;// 跳过最初的timestamp和表，库
            byte tmpByte = bytesSrc[i];// 获取操作类型
            types[cnt] = tmpByte;// 保存操作的记录
            i += 2;// 跳过 U|
            // System.err.println((char) tmpByte);
            // 更新数量最多
            if (tmpByte == Constants.UPDATE) {
                // id:1:1|1265277|1265277|score:1:0|65|221|
                // 跳过id:1:1|
                i += 7;
                // 1265277|1265277|score:1:0|65|221|
				/* 获取旧主键值 */
                curPK = 0;
                while ((tmpByte = bytesSrc[i++]) != Constants.SEP) {
                    curPK = curPK * 10 + tmpByte - Constants.ZERO;
                }
                changePK = 0;
				/*-----范围内主键---------*/
                while ((tmpByte = bytesSrc[i++]) != Constants.SEP) {
                    changePK = changePK * 10 + tmpByte - Constants.ZERO;
                }
                curPKs[cnt] = curPK;
                changePKs[cnt] = changePK;
                pos[cnt] = i;
                while ((bytesSrc[i++]) != Constants.SEPLINE) {
                }
                // 插入数量其次
                cnt++;// 读取下一条记录，统计这个块里面有多少变更记录
                continue;
            } else if (tmpByte == Constants.INSERT) {
                // 跳过id:1:1|NULL|
                i += 12;
                // 1|first_name:2:0|NULL|徐|last_name:2:0|NULL|依|sex:2:0|NULL|男|score:1:0|NULL|66|score1:1:0|
				/* 计算主键值 */
                curPK = 0;
                while ((tmpByte = bytesSrc[i++]) != Constants.SEP) {
                    curPK = curPK * 10 + (tmpByte - Constants.ZERO);
                }
                curPKs[cnt] = curPK;
                pos[cnt] = i;// first_name:2:0|NULL|徐|last_name:2:0|NULL|依|sex:2:0|NULL|男|score:1:0|NULL|66|
                i += 75;// 后续可以进行调整
                while ((bytesSrc[i++]) != Constants.SEPLINE) {
                }
                cnt++;// 读取下一条记录，统计这个块里面有多少变更记录
                continue;
            } else {
                i += 7;
                curPK = 0;
                while ((tmpByte = bytesSrc[i++]) != Constants.SEP) {
                    curPK = curPK * 10 + tmpByte - Constants.ZERO;
                }
                curPKs[cnt] = curPK;
				/* 删除可以直接跳跃一个大前缀 */
                i += 75;// 后续调整大小
                while ((bytesSrc[i++]) != Constants.SEPLINE) {
                }
                cnt++;// 读取下一条记录，统计这个块里面有多少变更记录
                continue;
            }

        }
        return cnt;
    }
}
